package hn.cch.kafka.controller;


import hn.cch.kafka.service.KafkaService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


@RestController
public class KafkaController {

    private final static Logger logger = LoggerFactory.getLogger(KafkaController.class);

    @Value(value = "${custom.kafka.topics:,}")
    private String topics;

    @Autowired
    private KafkaService kafkaService;

    @RequestMapping(value = "/send")
    public String send() {
        String topic = topics.split(",")[0];
        kafkaService.send(topic, "key", "[{\"name\":\"test\"},{\"name\":\"demo\"}]");
        return "OK";
    }

    @RequestMapping(value = "/subscribe/{topic}")
    public String subscribe(@PathVariable(value = "topic") String topic) {
        kafkaService.subscribe(topic);
        return "OK";
    }

}
